package com.jason.kafka;

import com.jason.utils.DateUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

/**
 * 消费者
 * 参照 https://blog.csdn.net/liu649983697/article/details/119456475
 * 批量消费
 * 同一消费者组内的消费者不能大于主题的分区数，否则有的消费者接收不到消息
 */
@Service
@Slf4j
public class ConsumerService {


//    @KafkaListener(topics = {"topicA"})
//    public void receive(String message, Acknowledgment ack){
//        log.info("---消费者接收主题:{},消息内容:{},当前时间:{}", "topicA", message, DateUtils.getDate());
//    }
//
//    @KafkaListener(topics = {"topicB"})
//    public void receive2(ConsumerRecord record, Acknowledgment ack){
//
//        log.info("---消费者接收主题:{},分区:{}，offset:{},消息内容:{},当前时间:{}", record.topic(),record.partition(),record.offset(),record.value(), DateUtils.getDate());
//    }

//    @KafkaListener(topics = {"topic_test"},groupId = "g1")
    public void receive3(ConsumerRecord record, Acknowledgment ack){

        log.info("---方法:receive3---消费者接收主题:{},分区:{}，offset:{},消息内容:{},当前时间:{},消费组:g1", record.topic(),record.partition(),record.offset(),record.value(), DateUtils.getDate());
        //ack.acknowledge() 消息确认，需设置手动提交。
        ack.acknowledge();
    }


//    @KafkaListener(topicPattern = "topic.*",groupId = "g1")
    public void receive4(ConsumerRecord record, Acknowledgment ack){

        log.info("---方法:receive4---消费者接收主题:{},分区:{}，offset:{},消息内容:{},当前时间:{},消费组:g1", record.topic(),record.partition(),record.offset(),record.value(), DateUtils.getDate());
    }

    @KafkaListener(id = "myContainerId", topicPattern = "topic_test",groupId = "g1",containerFactory = "myKafkaListenerContainerFactory",errorHandler = "consumerErrorHandler")
//    @KafkaListener(topicPattern = "topic_test",groupId = "g1")
    public void receive5(ConsumerRecord record, Acknowledgment ack){
        log.info("---方法:receive5---消费者接收主题:{},分区:{}，offset:{},消息内容:{},当前时间:{},消费组:g1", record.topic(),record.partition(),record.offset(),record.value(), DateUtils.getDate());
        ack.acknowledge();
    }

//    @KafkaListener(topicPattern = "topic_test",groupId = "g1")
    public void receive6(ConsumerRecord record, Acknowledgment ack){

        log.info("---方法:receive6---消费者接收主题:{},分区:{}，offset:{},消息内容:{},当前时间:{},消费组:g1", record.topic(),record.partition(),record.offset(),record.value(), DateUtils.getDate());
    }

//    @KafkaListener(topics = {"topic_test"},groupId = "g2")
    public void receive7(ConsumerRecord record, Acknowledgment ack){

        log.info("---方法:receive7---消费者接收主题:{},分区:{}，offset:{},消息内容:{},当前时间:{},消费组:g2", record.topic(),record.partition(),record.offset(),record.value(), DateUtils.getDate());
    }

//    @KafkaListener(groupId = "g2", topicPartitions = {
//            @TopicPartition(topic = "topic_test",partitions={"0","2"}, partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "60"))
//    })
//    public void receive8(ConsumerRecord record, Acknowledgment ack){
//
//        log.info("---方法:receive8---消费者接收主题:{},分区:{}，offset:{},消息内容:{},当前时间:{},消费组:g2", record.topic(),record.partition(),record.offset(),record.value(), DateUtils.getDate());
//    }
}
